{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Kusto Ingest\n",
    "\n",
    "The aim of this notebook is to provide an example of sourcing, processing and ingesting security data to a custom\n",
    "[Kusto aka Azure Data Explorer (ADE) cluster](https://docs.microsoft.com/en-us/azure/data-explorer/data-explorer-overview).\n",
    "\n",
    "Kusto/ADE is a fast and highly scalable data exploration service for log and telemetry data, hosted in Azure - and is used across Microsoft\n",
    "for analysing huge datasets of this sort.\n",
    "\n",
    "### Dependencies\n",
    "\n",
    "For this data ingestion notebook only, the following are additional requirements:\n",
    "\n",
    "```\n",
    "    > pip install azure-kusto-data azure-kusto-ingest\n",
    "```\n",
    "\n",
    "### Example Data\n",
    "\n",
    "We use the [Open Threat Research Forge Mordor Security Datasets](https://github.com/OTRF/Security-Datasets/), and the corresponding\n",
    "[MSTICPY Mordor Data Provider](https://msticpy.readthedocs.io/en/latest/data_acquisition/MordorData.html) to retrieve them.\n",
    "\n",
    "The plan here is to consider `Microsoft-Windows-Sysmon/Operational` data and create a set of database tables, each holding the \n",
    "events (merged from across different datasets) belonging to a specific event type. For more information on this source data,\n",
    "see [Sysinternals Sysmon](https://docs.microsoft.com/en-us/sysinternals/downloads/sysmon).\n",
    "\n",
    "After processing and uploading, we'll end up with a structure that looks like the following:\n",
    "\n",
    "```\n",
    "    cluster('msticpykustodemo')\n",
    "        .database('MsticPyKustoDemo')\n",
    "            .table('Event1')\n",
    "                ... set of Sysmon Process Creation events (Event Type 1)\n",
    "            .table('Event2')\n",
    "                ... set of Sysmon File Creation Time events (Event Type 2)\n",
    "            etc.\n",
    "```\n",
    "\n",
    "___See: [./Kusto-Analysis.ipynb](./Kusto-Analysis.ipynb) for details on data analysis, following on from the prep and ingestion here.___"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.kusto.data import KustoClient, KustoConnectionStringBuilder\n",
    "from azure.kusto.data.helpers import dataframe_from_result_table\n",
    "from azure.kusto.ingest import IngestionProperties, QueuedIngestClient\n",
    "import pandas as pd\n",
    "import re\n",
    "\n",
    "from msticpy.data import QueryProvider"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Kusto Config\n",
    "\n",
    "You'll need to provide Query and Ingest URIs for your own writable Kusto/ADE cluster. See \n",
    "[Quickstart: Create an Azure Data Explorer cluster and database](https://docs.microsoft.com/en-us/azure/data-explorer/create-cluster-database-portal?tabs=one-click-create-database)\n",
    "for guidance on how to set one of these up within your Azure subscription.\n",
    "\n",
    "When producing this example, I provisioned a dev/test cluster with the following configuration:\n",
    "\n",
    "```\n",
    "    Region = UK West\n",
    "    Availability zones = (none)\n",
    "    Cluster name = msticpykustodemo\n",
    "\n",
    "    Workload = Dev/test\n",
    "    Size = (none)\n",
    "    Compute = Dev(No SLA)_Standard_D11_v2\n",
    "\n",
    "    Scaling method = Manual scale\n",
    "    Instance count = 1\n",
    "\n",
    "    Streaming ingestion = Off\n",
    "    Enable purge = Off\n",
    "    Auto-Stop cluster = On\n",
    "\n",
    "    System-assigned identity = On\n",
    "    Double encryption = Off\n",
    "    User-assigned Identity = Off\n",
    "    User-assigned identities = (none)\n",
    "    Tenants permissions = My tenant only\n",
    "```\n",
    "\n",
    "<br>\n",
    "\n",
    "___Caution: watch out for costs! Azure Kusto/ADE clusters aren't cheap to run!___"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace these with your own values.\n",
    "\n",
    "KUSTO_CLUSTER_URI = 'https://msticpykustodemo.ukwest.kusto.windows.net'\n",
    "KUSTO_CLUSTER_INGEST_URI = 'https://ingest-msticpykustodemo.ukwest.kusto.windows.net'\n",
    "KUSTO_DATABASE = 'MsticPyKustoDemo'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Loading & Preparation\n",
    "\n",
    "Here we use the MSTICPY Mordor Query Provider to retrieve a number of example source datasets.\n",
    "\n",
    "We then split these up to create a series of Pandas DataFrames, each containing a\n",
    "single event type, for events from all sources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Retrieving Mitre data...\n",
      "Retrieving Mordor data...\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Downloading Mordor metadata: 100%|██████████| 96/96 [00:00<00:00, 2154.64 files/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[+] Retrieving Dataset \"atomic.windows.collection.host.msf_record_mic\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.cmd_lsass_memory_dumpert_syscalls\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.cmd_psexec_lsa_secrets_dump\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.cmd_sam_copy_esentutl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.covenant_dcsync_dcerpc_drsuapi_DsGetNCChanges\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_dcsync_dcerpc_drsuapi_DsGetNCChanges\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_mimikatz_backupkeys_dcerpc_smb_lsarpc\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_mimikatz_extract_keys\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_mimikatz_logonpasswords\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_mimikatz_lsadump_patch\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_mimikatz_sam_access\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_over_pth_patch_lsass\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_powerdump_sam_access\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_shell_reg_dump_sam\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_shell_rubeus_asktgt_createnetonly\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.empire_shell_rubeus_asktgt_ptt\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.psh_input_capture_promptforcreds\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.psh_lsass_memory_dump_comsvcs\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.psh_windows_vault_web_credentials\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.host.rdp_interactive_taskmanager_lsass_dump\"\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.network.covenant_dcsync_dcerpc_drsuapi_DsGetNCChanges\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.network.empire_dcsync_dcerpc_drsuapi_DsGetNCChanges\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.network.empire_mimikatz_backupkeys_dcerpc_smb_lsarpc\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.network.empire_shell_rubeus_asktgt_createnetonly\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.credential_access.network.empire_shell_rubeus_asktgt_ptt\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.cmd_bitsadmin_download_psh_script\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.cmd_mshta_javascript_getobject_sct\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.cmd_mshta_vbscript_execute_psh\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.cmd_netsh_fw_mod_open_ports\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.cmd_process_herpaderping_snippingtool\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.covenant_installutil\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.covenant_lolbin_wuauclt_createremotethread\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_dllinjection_LoadLibrary_CreateRemoteThread\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_launcher_sct_regsvr32\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_monologue_netntlm_downgrade\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_powerview_ldap_ntsecuritydescriptor\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_psinject_PEinjection\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_cmstp_execution_bypassuac\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_control_panel_execution\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_hh_local_html_payload\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_mavinject_dll_notepad\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_mshta_html_application_execution\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.psh_register_cimprovider_execute_dll\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.purplesharp_pe_injection_createremotethread\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.wmic_remote_xsl_jscript\"\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_enable_rdp.tar\"\n",
      "[!] Retrieving \"atomic.windows.defense_evasion.host.empire_enable_rdp.tar\" Failed [ File type not supported https://raw.githubusercontent.com/OTRF/Security-Datasets/master/datasets/atomic/windows/defense_evasion/host/empire_enable_rdp.tar.gz ]\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.host.empire_wdigest_downgrade.tar\"\n",
      "[!] Retrieving \"atomic.windows.defense_evasion.host.empire_wdigest_downgrade.tar\" Failed [ File type not supported https://raw.githubusercontent.com/OTRF/Security-Datasets/master/datasets/atomic/windows/defense_evasion/host/empire_wdigest_downgrade.tar.gz ]\n",
      "[+] Retrieving Dataset \"atomic.windows.defense_evasion.network.empire_powerview_ldap_ntsecuritydescriptor\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.cmd_discover_iexplorer_version_registry\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.cmd_seatbelt_group_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.covenant_getdomaingroup_ldap_searchrequest_domain_admins\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_find_localadmin_smb_svcctl_OpenSCManager\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_getsession_dcerpc_smb_srvsvc_NetSessEnum\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_shell_net_local_users\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_shell_net_localgroup_administrators\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_shell_rpc_samr_smb_group_domain_admins_standard_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.host.empire_shell_samr_EnumDomainUsers\"\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.network.covenant_getdomaingroup_ldap_searchrequest_domain_admins\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.network.empire_getsession_dcerpc_smb_srvsvc_NetSessEnum\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.network.empire_shell_rpc_samr_smb_group_domain_admins_standard_user\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.discovery.network.empire_shell_samr_EnumDomainUsers\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.execution.host.cmd_sharpview_pcre_net\"\n",
      "[+] Retrieving Dataset \"atomic.windows.execution.host.empire_launcher_vbs\"\n",
      "[+] Retrieving Dataset \"atomic.windows.execution.host.psh_powershell_httplistener\"\n",
      "[+] Retrieving Dataset \"atomic.windows.execution.host.psh_python_webserver\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.aadinternals_export_adfsdatabaseconfig_remotely\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_copy_smb_CreateRequest\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_dcom_executeexcel4macro_allowed\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_dcom_iertutil_dll_hijack\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_dcom_registerxll\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_psremoting_command\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sc_query_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sharpsc_create_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sharpsc_query_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sharpsc_start_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sharpsc_stop_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_sharpwmi_create_dcerpc_wmi\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_wmi_remote_event_subscription_ActiveScriptEventConsumers\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.covenant_wmi_wbemcomn_dll_hijack\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_dcom_shellwindows_stager\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_msbuild_dcerpc_wmi_smb\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_psexec_dcerpc_tcp_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_psremoting_stager\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_shell_dcerpc_smb_service_dll_hijack\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_smbexec_dcerpc_smb_svcctl\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_wmi_dcerpc_wmi_IWbemServices_ExecMethod\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.empire_wmic_add_user_backdoor\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2\"\n",
      "[!] Retrieving \"atomic.windows.lateral_movement.host.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2\" Failed [ QueryContainer object has no attribute windows.lateral_movement.host.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2 ]\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.purplesharp_ad_playbook_I\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.schtask_create\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.host.schtask_modification\"\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.aadinternals_export_adfsdatabaseconfig_remotely\"\n",
      "Cannot process files of type .pcapng\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_dcom_executeexcel4macro_allowed\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_dcom_iertutil_dll_hijack\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_dcom_registerxll\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_psremoting_command\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sc_query_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sharpsc_create_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sharpsc_query_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sharpsc_start_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sharpsc_stop_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_sharpwmi_create_dcerpc_wmi\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_wmi_remote_event_subscription_ActiveScriptEventConsumers\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.covenant_wmi_wbemcomn_dll_hijack\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_dcom_shellwindows_stager\"\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "        <style>\n",
       "            div.solid {border: thin solid black; padding:10px;}\n",
       "            p.title {background-color:Tomato; padding:5px;}\n",
       "            ul.circle {list-style-type: circle;}\n",
       "        </style>\n",
       "        <div class='solid'><h3><p class='title'>MsticpyUserError - Mordor download error</p></h3>Could not extract zip file for https://raw.githubusercontent.com/OTRF/Security-Datasets/master/datasets/atomic/windows/lateral_movement/network/empire_dcom_shellwindows_stager.zip.<br>File does not exist or is corrupt.<br><br>For more help on fixing this error see:<br><ul class='circle'><li><a href='https://msticpy.readthedocs.io/en/latest/data_acquisition/MordorData.html' target='_blank' rel='noopener noreferrer'>https://msticpy.readthedocs.io/en/latest/data_acquisition/MordorData.html</a></li></ul></div>"
      ],
      "text/plain": [
       "msticpy.common.exceptions.MsticpyUserError('Could not extract zip file for https://raw.githubusercontent.com/OTRF/Security-Datasets/master/datasets/atomic/windows/lateral_movement/network/empire_dcom_shellwindows_stager.zip.',\n",
       "                                           'File does not exist or is corrupt.')"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[!] Retrieving \"atomic.windows.lateral_movement.network.empire_dcom_shellwindows_stager\" Failed [ ('Mordor download error', 'Could not extract zip file for https://raw.githubusercontent.com/OTRF/Security-Datasets/master/datasets/atomic/windows/lateral_movement/network/empire_dcom_shellwindows_stager.zip.', 'File does not exist or is corrupt.', 'https://msticpy.readthedocs.io/en/latest/data_acquisition/MordorData.html') ]\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_msbuild_dcerpc_wmi_smb\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_psexec_dcerpc_tcp_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_psremoting_stager\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_shell_dcerpc_smb_service_dll_hijack\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_smbexec_dcerpc_smb_svcctl\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.empire_wmi_dcerpc_wmi_IWbemServices_ExecMethod\"\n",
      "Cannot process files of type .cap\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2\"\n",
      "[!] Retrieving \"atomic.windows.lateral_movement.network.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2\" Failed [ QueryContainer object has no attribute windows.lateral_movement.network.mimikatz_CVE-2020-1472_Unauthenticated_NetrServerAuthenticate2 ]\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.purplesharp_ad_playbook_I\"\n",
      "Cannot process files of type .cap\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.schtask_create\"\n",
      "Cannot process files of type .pcapng\n",
      "[+] Retrieving Dataset \"atomic.windows.lateral_movement.network.schtask_modification\"\n",
      "Cannot process files of type .pcapng\n",
      "[+] Retrieving Dataset \"atomic.windows.other.aptsimulator_cobaltstrike\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.cmd_userinitmprlogonscript_batch\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.empire_persistence_registry_modification_run_keys_elevated_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.empire_persistence_registry_modification_run_keys_standard_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.empire_schtasks_creation_execution_elevated_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.empire_schtasks_creation_standard_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.empire_wmi_local_event_subscriptions_elevated_user\"\n",
      "[+] Retrieving Dataset \"atomic.windows.persistence.host.proxylogon_ssrf_rce_poc\"\n",
      "[+] Retrieving Dataset \"atomic.windows.privilege_escalation.host.cmd_service_mod_fax\"\n",
      "[+] Retrieving Dataset \"atomic.windows.privilege_escalation.host.empire_uac_shellapi_fodhelper\"\n"
     ]
    }
   ],
   "source": [
    "mdr_data = QueryProvider(\"Mordor\", use_cached=True, save_folder='./mordor', silent=True)\n",
    "mdr_data.connect()\n",
    "\n",
    "events = {}\n",
    "\n",
    "for dataset in mdr_data.list_queries():\n",
    "    if dataset.startswith('atomic.windows'):\n",
    "        print('[+] Retrieving Dataset', f'\"{dataset}\"')\n",
    "        try:\n",
    "            df = getattr(mdr_data, dataset)(silent=True, use_cached=True)\n",
    "            if type(df) != pd.DataFrame:\n",
    "                continue\n",
    "            for i, r in df.iterrows():\n",
    "                row_entry = r.to_dict()\n",
    "                \n",
    "                # For demo purposes, we'll consider only Sysmon event logs.\n",
    "                if row_entry['Channel'] != 'Microsoft-Windows-Sysmon/Operational':\n",
    "                    continue\n",
    "\n",
    "                # Every entry should have a @timestamp:datetime field.\n",
    "                if '@timestamp' not in row_entry:\n",
    "                    if 'TimeCreated' not in row_entry:\n",
    "                        continue\n",
    "                    row_entry['@timestamp'] = row_entry['TimeCreated']\n",
    "                    del row_entry['TimeCreated']\n",
    "\n",
    "                for k, v in row_entry.items():\n",
    "                    # Try and parse datetimes, in various formats, to datetimes\n",
    "                    if type(v) != str:\n",
    "                        continue\n",
    "                    if re.match(r'^\\d{4}-\\d{2}-\\d{2}[ T]{1}\\d{2}:\\d{2}:\\d{2}(\\.\\d+){0,1}([+-][\\.\\d]{5}|Z){0,1}$', v):\n",
    "                        row_entry[k] = pd.to_datetime(v, utc=True)\n",
    "\n",
    "                # Normalise fieldnames - remove @, capitalise.\n",
    "                for k in list(row_entry.keys()):\n",
    "                    k_fixed = k\n",
    "                    if k_fixed[0] == '@': k_fixed = k_fixed[1:]\n",
    "                    if k_fixed[0].islower(): k_fixed = k_fixed[0].upper() + k_fixed[1:]\n",
    "                    if k != k_fixed:\n",
    "                        row_entry[k_fixed] = row_entry[k]\n",
    "                        del row_entry[k]\n",
    "\n",
    "                event_id = row_entry['EventID']\n",
    "                if not event_id in events:\n",
    "                    events[event_id] = []\n",
    "                events[event_id].append(row_entry)\n",
    "\n",
    "        except Exception as ex:\n",
    "            print('[!] Retrieving', f'\"{dataset}\"', 'Failed [', str(ex), ']')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[+] EventID: 1, #Records:1240, #Columns: 56\n",
      "[+] EventID: 2, #Records:532, #Columns: 39\n",
      "[+] EventID: 3, #Records:2592, #Columns: 51\n",
      "[+] EventID: 4, #Records:3, #Columns: 33\n",
      "[+] EventID: 5, #Records:1078, #Columns: 38\n",
      "[+] EventID: 6, #Records:1, #Columns: 35\n",
      "[+] EventID: 7, #Records:57139, #Columns: 49\n",
      "[+] EventID: 8, #Records:350, #Columns: 44\n",
      "[+] EventID: 9, #Records:1838, #Columns: 37\n",
      "[+] EventID: 10, #Records:103207, #Columns: 46\n",
      "[+] EventID: 11, #Records:5256, #Columns: 40\n",
      "[+] EventID: 12, #Records:114904, #Columns: 40\n",
      "[+] EventID: 13, #Records:50340, #Columns: 42\n",
      "[+] EventID: 14, #Records:1, #Columns: 34\n",
      "[+] EventID: 15, #Records:76, #Columns: 38\n",
      "[+] EventID: 17, #Records:256, #Columns: 38\n",
      "[+] EventID: 18, #Records:858, #Columns: 40\n",
      "[+] EventID: 19, #Records:3, #Columns: 37\n",
      "[+] EventID: 20, #Records:3, #Columns: 37\n",
      "[+] EventID: 21, #Records:3, #Columns: 36\n",
      "[+] EventID: 22, #Records:544, #Columns: 39\n",
      "[+] EventID: 23, #Records:1584, #Columns: 41\n",
      "[+] EventID: 24, #Records:17, #Columns: 39\n",
      "[+] EventID: 26, #Records:23, #Columns: 19\n",
      "[+] EventID: 255, #Records:6, #Columns: 32\n"
     ]
    }
   ],
   "source": [
    "for event_id in sorted(events.keys()):\n",
    "    df = pd.DataFrame(events[event_id])\n",
    "    df.sort_values('Timestamp', inplace=True)\n",
    "    df.reset_index(drop=True, inplace=True)\n",
    "\n",
    "    # Because the source dataframes (and hence rows for this dataframe) contained many event types,\n",
    "    # they have many irrelevant columns for this single event type.\n",
    "    # Here we attempt to filter out any columns that are all empty, dealing with different kinds of empty.\n",
    "\n",
    "    null_columns = []\n",
    "    for test_column in df.columns:\n",
    "        if df[~df[test_column].isnull() & (df[test_column]!='-')].empty:\n",
    "            null_columns.append(test_column)\n",
    "\n",
    "    df.drop(null_columns, axis=1, inplace=True)\n",
    "    events[event_id] = df\n",
    "\n",
    "    print('[+] EventID: ', event_id, ', #Records:', df.shape[0], ', #Columns: ', df.shape[1], sep='')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Kusto Ingestion\n",
    "\n",
    "First, we need to authenticate to Kusto/ADE. By default, we'll use \"az\" client auth - which requires\n",
    "that you run \"az login\" from a shell. See\n",
    "[Kusto connection strings](https://docs.microsoft.com/en-us/azure/data-explorer/kusto/api/connection-strings/kusto)\n",
    "for more details and alternative options.\n",
    "\n",
    "Next, we retrieve the existing database tables, then drop any with existing names that we intend to replace.\n",
    "\n",
    "Following that, we create a new table for each event type, then upload the DataFrame containing our records\n",
    "(with some type conversion as appropriate)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Use Azure CLI authentication: requires that you run \"az login\".\n",
    "\n",
    "kusto_client = KustoClient(KustoConnectionStringBuilder.with_az_cli_authentication(KUSTO_CLUSTER_URI))\n",
    "kusto_ingest_client = QueuedIngestClient(KustoConnectionStringBuilder.with_az_cli_authentication(KUSTO_CLUSTER_INGEST_URI))\n",
    "\n",
    "# Query existing tables\n",
    "\n",
    "db_tables = dataframe_from_result_table(kusto_client.execute_mgmt(KUSTO_DATABASE, '.show tables').primary_results[0])\n",
    "ingest_tables = [f'Event{k}' for k in events.keys()]\n",
    "\n",
    "# !!! Caution !!!\n",
    "# Drop any existing tables that match our new table names\n",
    "\n",
    "drop_tables = [tbl for tbl in ingest_tables if not db_tables[db_tables['TableName']==tbl].empty]\n",
    "if len(drop_tables):\n",
    "    db_tables = dataframe_from_result_table(kusto_client.execute_mgmt(KUSTO_DATABASE, f'.drop tables ({\",\".join(drop_tables)})').primary_results[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[+] Ingesting \"Event13\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event12\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event10\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event9\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event3\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event7\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event18\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event1\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event23\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event11\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event2\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event5\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event22\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event17\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event24\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event8\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event26\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event15\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event19\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event20\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event21\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event14\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event255\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event4\"...\n",
      "    IngestionStatus.QUEUED\n",
      "[+] Ingesting \"Event6\"...\n",
      "    IngestionStatus.QUEUED\n"
     ]
    }
   ],
   "source": [
    "pandas_kusto_type_mappings = {\n",
    "    'datetime64[ns, UTC]': 'datetime',\n",
    "    'object': 'string',\n",
    "    'float64': 'decimal',\n",
    "    'int64': 'long'\n",
    "}\n",
    "\n",
    "for event_key in events.keys():\n",
    "    tbl_name = f'Event{event_key}'\n",
    "    print('[+] Ingesting \"', tbl_name, '\"...', sep='')\n",
    "\n",
    "    event_columns = events[event_key].dtypes.to_dict()\n",
    "    event_columns = \", \".join([f'{k}:{pandas_kusto_type_mappings.get(str(v), str(v))}' for k, v in event_columns.items()])\n",
    "\n",
    "    kusto_client.execute_mgmt(KUSTO_DATABASE, f'.create table {tbl_name} ({event_columns})').primary_results[0]\n",
    "\n",
    "    # Ingest the DataFrame.\n",
    "\n",
    "    ingest_result = kusto_ingest_client.ingest_from_dataframe(\n",
    "        events[event_key],\n",
    "        ingestion_properties=IngestionProperties(\n",
    "            database=KUSTO_DATABASE,\n",
    "            table=tbl_name\n",
    "        )\n",
    "    )\n",
    "\n",
    "    print('   ', ingest_result.status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Done!\n",
    "\n",
    "You should soon be able to query and interact with the data in Kusto - see [Kusto-Analysis.ipynb](./Kusto-Analysis.ipynb) for details.\n",
    "\n",
    "Kusto make take some minutes to actually perform the ingest - and so data won't be available to query immediately!"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "970ffae9eb5607bf903c3b8348e40cded77164f57a6fb73fb45a1a063bddf4d8"
  },
  "kernelspec": {
   "display_name": "Python 3.10.1 ('.venv': venv)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
